package storageserver

import (
	"storageproto"
	"hashTbl"
	"rpc"
	"sort"
	"rand"
	"log"
	"fmt"
	"os"
	"time"
	"strings"
	"hash/fnv"
	"strconv"
	"json"
)

const (
	QUERY_COUNT_SECONDS = 5 // Request lease if this is the 3rd+
	QUERY_COUNT_THRESH = 3  // query in the last 5 seconds
	LEASE_SECONDS = 5       // Leases are valid for 5 seconds
	LEASE_GUARD_SECONDS = 1 // Servers add a short guard time
)


type cliList struct {
	List []storageproto.Client
}

// functions required by the sort.Interface interface
func (list cliList) Len() int{
	return len(list.List)
}

func (list cliList) Less(i, j int) bool{
	return list.List[i].NodeID < list.List[j].NodeID
}

func (list cliList) Swap(i, j int){
	list.List[i], list.List[j] = list.List[j], list.List[i]
}

// helper function for lists of clients
func (list cliList) Exists(obj storageproto.Client) bool {
	for idx := range list.List {
		cli := list.List[idx]
		if cli.HostPort == obj.HostPort && cli.NodeID == obj.NodeID {
			return true
		}
	}
	return false
}

type Storageserver struct {
	hashTbl *hashTbl.HashTable
	cache *hashTbl.HashTable
	myInfo storageproto.Client
	ready bool
	clients cliList
	rnd *rand.Rand
	//Ideally the servers would all connect to eachother once at the start, but they fail to connect if I try to connect
	//at the beggining
	//connections []*rpc.Client
}

type KeyLeases struct {
	LeaseClient storageproto.Client
	Expired int64
}

func NewStorageserver(master string, numnodes int, portnum int, nodeid uint32) *Storageserver {
	
	ss := &Storageserver{}
	ss.hashTbl = hashTbl.NewHashTable()
	ss.cache = hashTbl.NewHashTable()
	ss.rnd = rand.New(rand.NewSource(time.Nanoseconds()))
	if nodeid == 0 {
		//get a random nodeid
		nodeid = ss.rnd.Uint32()
	}
	ss.myInfo = storageproto.Client{fmt.Sprintf("localhost:%d",portnum),nodeid}
	ss.ready = false
	if (numnodes == 0) {
		// Not the master
		// get the list of all other servers from the master then connect to them.
		var masterCli *rpc.Client
		for true {
			//connect to the master, if the connection fails wait one second and try again.
			var err os.Error
			masterCli, err = rpc.DialHTTP("tcp", master)
			if err == nil {
				break
			}
			<- time.After(1000000000)
		}
		for true {
			//call Register on the master to get the list of clients, if the call fails (or the other clients aren't
			//ready) wait 1 second and try again.
			regArgs := storageproto.RegisterArgs{ss.myInfo}
			regReply := storageproto.RegisterReply{}
			err := masterCli.Call("StorageRPC.Register", regArgs, &regReply)
			if err == nil && regReply.Ready {
				masterCli.Close()
				ss.clients.List = regReply.Clients
				break
			}
			<- time.After(1000000000)
		}
		ss.readyServer()
		return ss
	}
	// master storage server, needs to register all others
	// create an array big enough to store all the client's info
	ss.clients.List = make([]storageproto.Client,1,numnodes)
	// insert the first element (master's info)
	ss.clients.List[0] = ss.myInfo
	if numnodes == 1 {
		// just the master server, only needs to register itself
		ss.readyServer()
	}
	return ss
}

func (ss *Storageserver) readyServer() {
	sort.Sort(ss.clients)
	/*ss.connections = make([]*rpc.Client,len(ss.clients.List),cap(ss.clients.List))
	for ii := 0; ii < len(ss.clients.List); ii++ {
		if ss.clients.List[ii].HostPort == ss.myInfo.HostPort && ss.clients.List[ii].NodeID == ss.myInfo.NodeID {
			ss.connections[ii] = nil
		} else {
			var err os.Error
			ss.connections[ii], err = rpc.DialHTTP("tcp", ss.clients.List[ii].HostPort)
			if err != nil {
				log.Fatalf("failed to connect to storageserver at %s", ss.clients.List[ii].HostPort)
			}
		}
	}*/
	//log.Printf("storageserver ready at %d\n", ss.myInfo.HostPort)
	ss.ready = true
}

func (ss *Storageserver) findServer(key string) int {
	substring := strings.Split(key,":")[0]
	h := fnv.New32()
	h.Write([]byte(substring))
	id := h.Sum32()
	clientIdx := sort.Search(len(ss.clients.List), func(i int) bool {return ss.clients.List[i].NodeID >= id}) % len(ss.clients.List)
	return clientIdx
}


// You might define here the functions that the locally-linked application
// logic can use to call things like Get, GetList, Put, etc.
// OR, you can have them access the local storage node
// by calling the RPC functions.  Your choice!

func (ss *Storageserver) addCacheQuery(key string) int {

	value, status := ss.cache.GET (fmt.Sprintf("%s:queries", key))
	timestamp := time.Seconds()

	var times []string

	if status == storageproto.OK {
		times = strings.Split(value, ",")
	}

	var startIndex int

	if len(times) == QUERY_COUNT_THRESH {
		startIndex = 1
	} else {
		startIndex = 0
	}

	var newVal string
	count := 0

	for i := startIndex; i < len(times); i++ {
		timeInt, err := strconv.Atoi64(times[i])
		if err != nil {
			return -1
		}
		if timeInt + QUERY_COUNT_SECONDS > time.Seconds() {
			if count == 0 {
				newVal = times[i]
				count++
			} else {
				newVal = strings.Join([]string{newVal,times[i]}, ",")
				count++
			}
		}
	}

        if count == 0 {
                newVal = fmt.Sprintf("%d", timestamp)
                count++
        } else {
                newVal = strings.Join([]string{newVal,fmt.Sprintf("%d", timestamp)}, ",")
		count++
        }

	ss.cache.PUT (fmt.Sprintf("%s:queries", key), newVal)

	return count
}

func (ss *Storageserver) checkCache(key string) bool {

	value, status := ss.cache.GET (fmt.Sprintf("%s:leases", key))

	if status == storageproto.OK {
		timePassed, err := strconv.Atoi64(value)

		if err != nil {
			return false
		}
		if timePassed > time.Seconds() {
			return true
		}
	}

	return false
}

func (ss *Storageserver) GET(key string) (string, int) {

	args := storageproto.GetArgs{key, false, storageproto.Client{}}
	reply := storageproto.GetReply{}

	err := ss.GetRPC(&args, &reply)
	if err != nil {
		return "", storageproto.EKEYNOTFOUND
	}
	
	return reply.Value, reply.Status
}

func (ss *Storageserver) PUT(key string, val string) int {
	
	args := storageproto.PutArgs{key,val}
	reply := storageproto.PutReply{}

	err := ss.PutRPC(&args, &reply)
	if err != nil {
		return storageproto.EPUTFAILED
	}

	return reply.Status
}

func (ss *Storageserver) GETLIST(key string) ([]string, int) {

    args := storageproto.GetArgs{key, false, storageproto.Client{}}
	reply := storageproto.GetListReply{}

	err := ss.GetListRPC(&args, &reply)
	if err != nil {
		return nil, storageproto.EKEYNOTFOUND
	}

	return reply.Value, reply.Status
}

func (ss *Storageserver) AddToList(key string, val string) int {

	args := storageproto.PutArgs{key,val}
	reply := storageproto.PutReply{}

	err := ss.AppendToListRPC(&args, &reply)
	if err != nil {
		return storageproto.EPUTFAILED
	}
	
	return reply.Status
}

func (ss *Storageserver) RemoveFromList(key string, val string) int {
	
	args := storageproto.PutArgs{key,val}
	reply := storageproto.PutReply{}

	err := ss.RemoveFromListRPC( &args, &reply)
	if err != nil {
		return storageproto.EKEYNOTFOUND
	}
	return reply.Status
}


// RPC-able interfaces, bridged via StorageRPC.
// These should do something! :-)

func (ss *Storageserver) RegisterRPC(args *storageproto.RegisterArgs, reply *storageproto.RegisterReply) os.Error {
	//if not all clients are registered, check to see if this is an unregistered client
	if !ss.ready {
		reply.Ready = false
		newcli := args.ClientInfo
		if !ss.clients.Exists(newcli){
			//add the new client to the list
			curlength := len(ss.clients.List)
			ss.clients.List = ss.clients.List[0:curlength+1]
			ss.clients.List[curlength] = newcli
			//check if that was the last client reqired
			if len(ss.clients.List) == cap(ss.clients.List) {
				ss.readyServer()
			}
		}
	}
	//if all clients are registered, return the list of all clients.
	if ss.ready {
		reply.Ready = true
		reply.Clients = ss.clients.List
	}
	return nil
}

func (ss *Storageserver) addLease(key string, leaseClient storageproto.Client) {
        // if available { 
        str, _ := json.Marshal(KeyLeases{leaseClient, time.Seconds() + int64(LEASE_SECONDS)})
        ss.cache.AddToList(fmt.Sprintf("%s:LeaseGranted", key), string(str))
	// }
}

func (ss *Storageserver) clearLeases(key string) int{
	leases, status := ss.cache.GETLIST(fmt.Sprintf("%s:LeaseGranted",key))
	//log.Printf("clear %d leases of %s",len(leases),key)
	if status == storageproto.EKEYNOTFOUND {
		return storageproto.OK
	}
	
	for _, val := range leases {
		var curLease KeyLeases
		json.Unmarshal([]byte(val), &curLease)
		log.Printf("revoke lease at %s",curLease.LeaseClient.HostPort)
		log.Printf("expired: %d, now: %d",curLease.Expired,time.Seconds())
		if curLease.Expired > time.Seconds(){ // + LEASE_GUARD_SECONDS {
			//connect to the rpc server
			log.Printf("revoke: connect")
			cli, err := rpc.DialHTTP("tcp", curLease.LeaseClient.HostPort)
			if err == nil {
				args := storageproto.RevokeLeaseArgs{key}
				reply := storageproto.RevokeLeaseReply{}
				log.Printf("revoke: call rpc")
				err = cli.Call("StorageRPC.RevokeLease", &args, &reply)
				if err == nil {

					//Close the connection
					cli.Close()

					// If there was an error for some reason, wait for the lease to expire
					if reply.Status == storageproto.OK {
						continue
					}
				}
			}
			//log.Printf("revokeRPC failed, waiting...")
			<- time.After(1000000000 * (time.Seconds() + LEASE_GUARD_SECONDS - curLease.Expired))
		}
		ss.cache.RemoveFromList(fmt.Sprintf("%s:LeaseGranted",key), val)
	}
	return storageproto.OK
}

func (ss *Storageserver) GetRPC(args *storageproto.GetArgs, reply *storageproto.GetReply) os.Error {
	
	clientId := ss.findServer(args.Key)
	
	//check to see if the request should be forwarded to a different server (or returned from a cached value)
	if ss.clients.List[clientId].HostPort != ss.myInfo.HostPort {
		
		//check to see if the value is in the local cache
		if ss.checkCache(args.Key) {
			log.Printf("return cached value for %s",args.Key)
			reply.Value,reply.Status = ss.cache.GET(args.Key)
			reply.Lease.Granted = false
			return nil
		}

		//check to see if the request is frequent enough to warrent caching
		wantsLease := (ss.addCacheQuery(args.Key) >= QUERY_COUNT_THRESH)
		
		if wantsLease {
			args.WantLease = true
			args.LeaseClient = ss.myInfo
		}

		//connect to the rpc server
		cli, err := rpc.DialHTTP("tcp", ss.clients.List[clientId].HostPort)
		if err != nil {
			reply.Status = storageproto.EKEYNOTFOUND
			return err
		}

		//make the rpc call
		err = cli.Call("StorageRPC.Get", args, reply)
		
		if err != nil {
			reply.Status = storageproto.EKEYNOTFOUND
			return err
		}
		//Close the connection
		cli.Close()
		
		//if a lease was requested (and granted) put the value in the local cache
		if wantsLease && reply.Lease.Granted {
			ss.cache.PUT(args.Key,reply.Value)
			ss.cache.PUT (fmt.Sprintf("%s:leases", args.Key), fmt.Sprintf("%d", time.Seconds() + int64(reply.Lease.ValidSeconds)))
		}

		return nil
	}
	//handle the request

	reply.Value, reply.Status = ss.hashTbl.GET(args.Key)

	// Check for lease request
	clearingList,_ := ss.cache.GETLIST(fmt.Sprintf("%s:clearingLeases",args.Key))
	if args.WantLease && (clearingList == nil || len(clearingList) == 0){
		ss.addLease(args.Key, args.LeaseClient)
		reply.Lease = storageproto.LeaseStruct{true, LEASE_SECONDS}
	} else {
		reply.Lease = storageproto.LeaseStruct{false, 0}
	}
	return nil
}

func (ss *Storageserver) GetListRPC(args *storageproto.GetArgs, reply *storageproto.GetListReply) os.Error {

	clientId := ss.findServer(args.Key)
	
	//check to see if the request should be forwarded to a different server (or returned from a cached value)
	if ss.clients.List[clientId].HostPort != ss.myInfo.HostPort {
		
		//check to see if the value is in the local cache
		if ss.checkCache(args.Key) {
			reply.Value,reply.Status = ss.cache.GETLIST(args.Key)
			reply.Lease.Granted = false
			return nil
		}

		//check to see if the request is frequent enough to warrent caching
		wantsLease := (ss.addCacheQuery(args.Key) >= QUERY_COUNT_THRESH)
		
		if wantsLease {
			args.WantLease = true
			args.LeaseClient = ss.myInfo
		}

		//connect to the rpc server
		cli, err := rpc.DialHTTP("tcp", ss.clients.List[clientId].HostPort)
		if err != nil {
			reply.Status = storageproto.EKEYNOTFOUND
			return err
		}

		//make the rpc call
		err = cli.Call("StorageRPC.GetList", args, reply)
		
		if err != nil {
			reply.Status = storageproto.EKEYNOTFOUND
			return err
		}
		//Close the connection
		cli.Close()
		
		//if a lease was requested (and granted) put the value in the local cache
		if wantsLease && reply.Lease.Granted {
			ss.cache.PUTLIST(args.Key,reply.Value)
			ss.cache.PUT (fmt.Sprintf("%s:leases", args.Key), fmt.Sprintf("%d", time.Seconds() + int64(reply.Lease.ValidSeconds)))
		}

		return nil
	}
	//handle the request

	reply.Value, reply.Status = ss.hashTbl.GETLIST(args.Key)

	// Check for lease request
	clearingList,_ := ss.cache.GETLIST(fmt.Sprintf("%s:clearingLeases",args.Key))
	if args.WantLease && (clearingList == nil || len(clearingList) == 0){
		ss.addLease(args.Key, args.LeaseClient)
		reply.Lease = storageproto.LeaseStruct{true, LEASE_SECONDS}
	} else {
		reply.Lease = storageproto.LeaseStruct{false, 0}
	}
	return nil
}

func (ss *Storageserver) PutRPC(args *storageproto.PutArgs, reply *storageproto.PutReply) os.Error {
	
	clientId := ss.findServer(args.Key)
	
	//check if the rpc should be forwarded to a different server
	if ss.clients.List[clientId].HostPort != ss.myInfo.HostPort {

		//connect to the rpc server
		cli, err := rpc.DialHTTP("tcp", ss.clients.List[clientId].HostPort)
		if err != nil {
			reply.Status = storageproto.EPUTFAILED
			return err
		}

		err = cli.Call("StorageRPC.Put", args, &reply)
		if err != nil {
			reply.Status = storageproto.EPUTFAILED
			return err
		}
		//Close the connection
		cli.Close()
	
		return nil
	}
	//handle the request

	// stop giving out leases
	//get a random opperation ID
	opId := ss.rnd.Uint32()
	ss.cache.AddToList(fmt.Sprintf("%s:clearingLeases",args.Key),fmt.Sprintf("opperation:%d",opId))
	
	// clear outstanding leases
	status := ss.clearLeases(args.Key)

	//log.Printf("Return code for releasing lease was: %d\n", status)

	if status != storageproto.OK {
		reply.Status = status
	} else {
		reply.Status = ss.hashTbl.PUT(args.Key,args.Value)
	}

	// signal leaseable again
	ss.cache.RemoveFromList(fmt.Sprintf("%s:clearingLeases",args.Key),fmt.Sprintf("opperation:%d",opId))
	return nil
}

func (ss *Storageserver) AppendToListRPC(args *storageproto.PutArgs, reply *storageproto.PutReply) os.Error {
	
	clientId := ss.findServer(args.Key)

	//check if the rpc should be forwarded to a different server
	if ss.clients.List[clientId].HostPort != ss.myInfo.HostPort {

		//connect to the rpc server
		cli, err := rpc.DialHTTP("tcp", ss.clients.List[clientId].HostPort)
		if err != nil {
			reply.Status = storageproto.EPUTFAILED
			return err
		}

		err = cli.Call("StorageRPC.AppendToList", args, &reply)
		if err != nil {
			reply.Status = storageproto.EPUTFAILED
			return err
		}
		//Close the connection
		cli.Close()
		
		return nil
	}
	//handle the request
	
	// stop giving out leases
	//get a random opperation ID
	opId := ss.rnd.Uint32()

	ss.cache.AddToList(fmt.Sprintf("%s:clearingLeases",args.Key),fmt.Sprintf("opperation:%d",opId))
	
	// clear outstanding leases
	status := ss.clearLeases(args.Key)
	if status != storageproto.OK {
		reply.Status = status
	} else {
		reply.Status = ss.hashTbl.AddToList(args.Key, args.Value)
	} 

	//signal leaseable again
	ss.cache.RemoveFromList(fmt.Sprintf("%s:clearingLeases",args.Key),fmt.Sprintf("opperation:%d",opId))
	return nil
}

func (ss *Storageserver) RemoveFromListRPC(args *storageproto.PutArgs, reply *storageproto.PutReply) os.Error {
	
	clientId := ss.findServer(args.Key)

	//check if the rpc should be forwarded to a different server
	if ss.clients.List[clientId].HostPort != ss.myInfo.HostPort {
		//connect to the rpc server
		cli, err := rpc.DialHTTP("tcp", ss.clients.List[clientId].HostPort)
		if err != nil {
			reply.Status = storageproto.EKEYNOTFOUND
			return err
		}
	
		err = cli.Call("StorageRPC.RemoveFromList", args, reply)
		if err != nil {
			reply.Status = storageproto.EKEYNOTFOUND
			return err
		}
		//Close the connection
		cli.Close()
		return nil
	}
	//handle the request

	// stop giving out leases
	// get a random opperation ID
	opId := ss.rnd.Uint32()
	
	ss.cache.AddToList(fmt.Sprintf("%s:clearingLeases",args.Key),fmt.Sprintf("opperation:%d",opId))
	
	// clear outstanding leases
	status := ss.clearLeases(args.Key)
	if status != storageproto.OK {
		reply.Status = status
	} else {
		reply.Status = ss.hashTbl.RemoveFromList(args.Key, args.Value)
	}

	// signal leaseable again
	ss.cache.RemoveFromList(fmt.Sprintf("%s:clearingLeases",args.Key),fmt.Sprintf("opperation:%d",opId))
	return nil
}

func (ss *Storageserver) RevokeLeaseRPC(args *storageproto.RevokeLeaseArgs, reply *storageproto.RevokeLeaseReply) os.Error {
	log.Printf("revoke lease for %s",args.Key)
	reply.Status = ss.cache.PUT(fmt.Sprintf("%s:leases", args.Key), "0")
	return nil
}
